package icu.wdm.flinkcdc.mysql;

import cn.hutool.core.date.DateUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.MySqlSourceBuilder;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestartStrategyOptions;
import org.apache.flink.configuration.StateBackendOptions;
import org.apache.flink.core.execution.CheckpointingMode;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.catalog.ObjectPath;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.File;
import java.io.IOException;
import java.time.Duration;
import java.util.Properties;

@Component
public class MySqlListenerStart {

    @Autowired
    private MySqlSourceProperties config;

    @EventListener
    public void start(ContextRefreshedEvent event) throws IOException {
        if (event.getApplicationContext().getParent() == null) {

            //1.获取flink 执行环境
            Configuration config1 = new Configuration();
            config1.set(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay");
            config1.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 3); // 尝试重启次数
            config1.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY, Duration.ofSeconds(10)); // 延时
            config1.set(StateBackendOptions.STATE_BACKEND, "hashmap");

            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config1);
            env.setParallelism(4);
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

            Properties prop = new Properties();
            prop.setProperty("useSSL", "false");
            prop.setProperty("characterEncoding", "UTF-8");
            prop.setProperty("useUnicode", "true");
            prop.setProperty("tinyInt1isBit", "false");
            prop.setProperty("allowPublicKeyRetrieval", "true");

            Properties debeziumProperties = new Properties();
            debeziumProperties.setProperty("converters", "dateConverters");
            debeziumProperties.setProperty("dateConverters.type", "icu.wdm.flinkcdc.mysql.DateTimeConverter");
            debeziumProperties.setProperty("dateConverters.format.date", "yyyy-MM-dd");
            debeziumProperties.setProperty("dateConverters.format.time", "HH:mm:ss");
            debeziumProperties.setProperty("dateConverters.format.datetime", "yyyy-MM-dd HH:mm:ss");
            debeziumProperties.setProperty("dateConverters.format.timestamp", "yyyy-MM-dd HH:mm:ss");
            debeziumProperties.setProperty("dateConverters.format.timestamp.zone", "UTC+8");


            //通过FlinkCDC构建SourceFunction
            MySqlSourceBuilder<String> builder = MySqlSource.<String>builder()
                    .hostname(config.getHost())
                    .port(config.getPort())
                    .username(config.getUsername())
                    .password(config.getPassword())
                    .serverTimeZone("Asia/Shanghai")
                    .jdbcProperties(prop)
                    .debeziumProperties(debeziumProperties)
                    .includeSchemaChanges(true)
                    .databaseList(config.getDatabaseList())
                    .tableList(config.getTableList())
                    .chunkKeyColumn(ObjectPath.fromString("paper.pubmed_click"), "pubmed_data_id")
                    .chunkKeyColumn(ObjectPath.fromString("paper.user_journal"), "user_id")
                    .deserializer(new JsonDebeziumDeserializationSchema())
                    .splitSize(10000) // 调整分片大小
                    .fetchSize(10000);// 调整抓取大小

            //设置同步模式
            StartupOptions startupOptions;
            if (StringUtils.hasText(config.getStartDate())) {
                startupOptions = StartupOptions.timestamp(DateUtil.parseDate(config.getStartDate()).getTime());
            } else if ("initial".equals(config.getStartUpMode())) {
                startupOptions = StartupOptions.initial(); // 全量同步
            } else {
                startupOptions = StartupOptions.latest(); // 增量同步
            }
            builder.startupOptions(startupOptions);


            DataStreamSource<String> mySqlDS = env.fromSource(builder.build(), WatermarkStrategy.noWatermarks(), config.getName());
            mySqlDS.addSink(new MysqlSink());

            try {
                env.execute("mysqlSynchronization");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    }
}
